package main.test.GraphFramesAPI

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{Column, DataFrame, SQLContext, functions}
import org.graphframes.GraphFrame
import org.graphframes.lib.AggregateMessages

object MessagePassingViaAggMess {
  def main(args: Array[String]): Unit = {
    //  def constructGraph: GraphFrame = {
    //    The following example demonstrates how
    //    to create a GraphFrame from vertex and edge DataFrame
    val sparkConf = new SparkConf()
    sparkConf.setAppName("createGraph").setMaster("local[*]")
    val sc = new SparkContext(sparkConf)
    val sqlContext = new SQLContext(sc)
    val v = sqlContext.createDataFrame(List(
      ("a", "Alice", 34),
      ("b", "Bob", 36),
      ("c", "Charlie", 30),
      ("d", "David", 29),
      ("e", "Esther", 32),
      ("f", "Fanny", 36),
      ("g", "Gabby", 60)
    )).toDF("id", "name", "age")

    // Edge DataFrame
    val e = sqlContext.createDataFrame(List(
      ("a", "b", "friend"),
      ("b", "c", "follow"),
      ("c", "b", "follow"),
      ("f", "c", "follow"),
      ("e", "f", "follow"),
      ("e", "d", "friend"),
      ("d", "a", "friend"),
      ("a", "e", "friend")
    )).toDF("src", "dst", "relationship")


    // Create a GraphFrame
    val g = GraphFrame(v, e)


    /**
     * Like Graphx, GraphFrame provides primitives for developing graph algorithms.
     * The key two components are:
     * 1  aggregateMessages: send messages between vertices, and aggregate messages for each
     * vertex. GraphFrames analogously to the GraphX api.
     * 2 joins: Join message aggregates with original graph. GF rely on DF joins,
     * which provide the full functionality of GraphX joins.
     * Below show how to use aggregateMessages to compute the sum of the ages of adjacent users.
     *
     */


    // We will use AggregateMessages utilities later, so name it "AM" for short.
    val AM = AggregateMessages

    // For each user, sum the ages of the adjacent users.

    val msgToSrc: Column = AM.dst("age")
    val msgToDst: Column = AM.src("age")

    val agg: DataFrame = g.aggregateMessages
        .sendToSrc(msgToSrc)
        .sendToDst(msgToDst)
      // error， sum()这个地方不能识别，说明这个地方没有sum这个函数
      //该代码来源于官方文档说明。
      .agg(functions.sum(AM.msg).as("summedAges"))
//        .agg( AM.msg as("summedAges") )     }
    agg.show(truncate = false)
  }

}
